package org.pizza.didi.redis;

import org.pizza.did.api.WorkerIdDistributor;

import org.pizza.did.event.WorkIdChangeEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.util.StringUtils;

import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * redis分配器
 */
public class RedisDistributor implements WorkerIdDistributor, ApplicationContextAware {
    private static final Logger log = LoggerFactory.getLogger(RedisDistributor.class);
    public static final String ID_PREFIX = "did:";
    /**
     * 应用名
     */
    private String applicationName;
    /**
     * 应该所在IP
     */
    private String ip;
    /**
     * 应用端口
     */
    private String port;
    /**
     * 当前节点 => ip:port
     */
    private String currentNode;
    /**
     * 最大maxId
     */
    private long maxId;
    /**
     * 获取到的workerId
     */
    private long workerId;
    /**
     * 当前节点槽
     */
    private String currentNodeSlot;
    /**
     * 最后的更新时间
     */
    private long lastUpdateTime;
    /**
     * 监听状态
     */
    private volatile boolean monitoring = false;

    private StringRedisTemplate stringRedisTemplate;
    private ApplicationContext applicationContext;



    public RedisDistributor(String applicationName, String ip, String port, long maxId, StringRedisTemplate stringRedisTemplate) {
        this.applicationName = applicationName;
        this.ip = ip;
        this.port = port;
        this.currentNode = ip + ":" + port;
        this.maxId = maxId;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Override
    public boolean init() {
        return true;
    }

    /**
     * 进行分配
     * @return
     */
    private boolean doDistribute() {
        for (int i = 0; i <= maxId; i++) {
            //每个slot被一个服务节点占用
            String slot = ID_PREFIX + applicationName + ":slot:" + i;
            //获取slot的拥有者
            BoundValueOperations<String, String> operations = stringRedisTemplate.boundValueOps(slot);
            String holder = operations.get();
            //没有拥有者
            if (StringUtils.isEmpty(holder)) {
                Boolean absent = operations.setIfAbsent(this.currentNode, 300, TimeUnit.SECONDS);
                if (null == absent || !absent) {
                    log.warn("RedisDistributor 争抢 [{}] 失败，疑似并发  ip:{} port:{}", slot, ip, port);
                    break;
                }
                this.currentNodeSlot = slot;
                this.workerId = i;
                log.info("RedisDistributor 争抢 [{}] 成功 ip:{} port:{} workId:{}", slot, ip, port, workerId);
                return true;
            }
            //有拥有者且为当前服务(短时间掉线)，继续分配给当前服务
            if (holder.equals(currentNode)) {
                this.currentNodeSlot = slot;
                this.workerId = i;
                log.info("RedisDistributor 重新持有 [{}] 成功 ip:{} port:{} workId:{}", slot, ip, port, workerId);
                return true;
            }
        }
        //超过最大workId还没有获取成功
        return false;
    }

    @Override
    public long distribute() {
        if (doDistribute()) {
            if (!monitoring) {
                this.monitoring = true;
                this.scheduledUploadData();
            }
            return this.workerId;
        }
        throw new RuntimeException("分配workId失败");
    }

    private void scheduledUploadData() {
        Executors.newSingleThreadScheduledExecutor(r -> {
            Thread thread = new Thread(r, "RedisDistributor-");
            thread.setDaemon(true);
            return thread;
        })
                //每60s上报数据
                .scheduleWithFixedDelay(this::upload, 0L, 60, TimeUnit.SECONDS);
    }

    /**
     * 上报保活
     */
    private void upload() {
        //获取当前节点槽
        BoundValueOperations<String, String> operations = this.stringRedisTemplate.boundValueOps(this.currentNodeSlot);
        //拥有者
        String holder = operations.get();
        //如果没有持有者了（被手动清空了?维护超时了？）
        if (StringUtils.isEmpty(holder)) {
            Boolean absent = operations.setIfAbsent(this.currentNode, 300, TimeUnit.SECONDS);
            if (null == absent || !absent) {
                //定时任务会重新监听
                log.warn("RedisDistributor 争抢 [{}] 失败，疑似并发  ip:{} port:{}", currentNodeSlot, ip, port);
            }
        }
        //持有者就是当前服务（继续维护续约）
        else if (holder.equals(currentNode)) {
            Boolean expire = operations.expire(300, TimeUnit.SECONDS);
            //更新过期时间失败（被手动清空了?通信异常？）
            if (null == expire || !expire) {
                //定时任务会重新监听
                log.warn("RedisDistributor  [{}] 更新失败 ip:{} port:{}", currentNodeSlot, ip, port);
            }
        }
        //有持有者，但不是当前服务（被手动清空了?维护超时了？定时任务监控的时间差被别的服务抢走了~~）
        else {
            log.error("RedisDistributor [{}] 被 {} 节点抢走了，避免ID生成冲突，发送事件重新获取", currentNodeSlot, holder);
            applicationContext.publishEvent(new WorkIdChangeEvent(this));
        }
    }

    public String getApplicationName() {
        return applicationName;
    }

    public void setApplicationName(String applicationName) {
        this.applicationName = applicationName;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public String getPort() {
        return port;
    }

    public void setPort(String port) {
        this.port = port;
    }

    public long getMaxId() {
        return maxId;
    }

    public void setMaxId(long maxId) {
        this.maxId = maxId;
    }

    public StringRedisTemplate getStringRedisTemplate() {
        return stringRedisTemplate;
    }

    public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    public long getWorkerId() {
        return workerId;
    }

    public void setWorkerId(long workerId) {
        this.workerId = workerId;
    }

    public String getCurrentNodeSlot() {
        return currentNodeSlot;
    }

    public void setCurrentNodeSlot(String currentNodeSlot) {
        this.currentNodeSlot = currentNodeSlot;
    }

    public long getLastUpdateTime() {
        return lastUpdateTime;
    }

    public void setLastUpdateTime(long lastUpdateTime) {
        this.lastUpdateTime = lastUpdateTime;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
